package com.hlz.flink.chapter06;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import java.util.Calendar;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * 在自定义数据源中发射水位线
 *
 * @author Hongliang Zhu
 * @create 2023-06-23 11:00
 */
public class EmitWatermarkInSourceFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.addSource(new ClickSourceWatermark()).print("EmitWatermarkInSourceFunction");
        env.execute();
    }

    public static class ClickSourceWatermark implements SourceFunction<Stock> {
        private boolean running = true;

        @Override
        public void run(SourceContext<Stock> ctx) throws Exception {

            Random random = new Random();
            double[] price = {23.6, 34, 5, 12, 4, 67.2, 32.3, 15, 9};


            while (running) {
                long currTs = Calendar.getInstance().getTimeInMillis();
                String uuid = UUID.randomUUID().toString();

                double tosckPrice = price[random.nextInt(price.length)];
                Stock udfStock = new Stock("stock_" + uuid, tosckPrice, "UDFStock", currTs);
                // 使用 collectWithTimestamp 方法将数据发送出去，并指明数据中的时间戳的字段
                ctx.collectWithTimestamp(udfStock, udfStock.getTimeStamp());
                // 发送水位线
                ctx.emitWatermark(new Watermark(udfStock.getTimeStamp() - 1L));
                TimeUnit.SECONDS.sleep(1);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}
